package site.teamo.biu.flink.sdk.job;


import cn.hutool.core.collection.CollectionUtil;

import java.util.*;
import java.util.function.Consumer;


/**
 * @author haocongshun
 * @date 2023/07/26 11:04:31
 */
public class Pipeline {

    private final String root;

    private final DAG<String> dag;

    private final Map<String, JobNode> jobNodeMap;
    private final Map<String, String> parentMapping;

    public Pipeline(List<JobNodeInfo> nodeInfos, JobNodeInfo root) {
        this.dag = new DAG<>();
        this.jobNodeMap = new HashMap<>();
        this.parentMapping = new HashMap<>();
        this.root = root.getName();
        for (JobNodeInfo nodeInfo : nodeInfos) {
            this.dag.addNode(nodeInfo.getName());
            this.jobNodeMap.put(nodeInfo.getName(), new JobNode(nodeInfo));
        }
        for (JobNodeInfo nodeInfo : nodeInfos) {
            if (CollectionUtil.isEmpty(nodeInfo.getNextNode())) {
                continue;
            }
            for (String s : nodeInfo.getNextNode()) {
                this.dag.addEdge(nodeInfo.getName(), s);
            }
        }
        this.jobNodeMap.keySet()
                .forEach(name -> {
                    List<String> parents = this.dag.getParents(name);
                    if (CollectionUtil.isEmpty(parents)) {
                        return;
                    }
                    this.parentMapping.put(name, parents.get(0));
                });

    }

    public String toCode() {
        StringBuilder sb = new StringBuilder();
        this.dag.breadthFirstTraversal(this.root, name -> {
            sb.append(this.jobNodeMap.get(name).toCode(parentMapping.get(name)));
            sb.append("\n");
        });
        return sb.toString();
    }

    private static class DAG<T> {
        private final Map<T, DAGNode<T>> nodes;
        private final Set<DAGNode<T>> visited;

        public DAG() {
            this.nodes = new HashMap<>();
            this.visited = new HashSet<>();
        }

        public void addNode(T data) {
            if (!nodes.containsKey(data)) {
                nodes.put(data, new DAGNode<>(data));
            }
        }

        public void addEdge(T from, T to) {
            DAGNode<T> fromNode = nodes.get(from);
            DAGNode<T> toNode = nodes.get(to);

            if (fromNode != null && toNode != null) {
                if (hasCycle(fromNode, toNode)) {
                    throw new IllegalArgumentException("Adding this edge would create a cycle.");
                }
                fromNode.addChild(toNode);
            }
        }

        private boolean hasCycle(DAGNode<T> fromNode, DAGNode<T> toNode) {
            visited.clear();
            return hasCycleDFS(fromNode, toNode);
        }

        private boolean hasCycleDFS(DAGNode<T> current, DAGNode<T> target) {
            if (current == target) {
                return true;
            }

            if (visited.contains(current)) {
                return false;
            }

            visited.add(current);

            for (DAGNode<T> child : current.getChildren()) {
                if (hasCycleDFS(child, target)) {
                    return true;
                }
            }

            visited.remove(current);
            return false;
        }

        public List<T> getParents(T childData) {
            DAGNode<T> childNode = nodes.get(childData);
            List<T> parents = new ArrayList<>();
            if (childNode == null) {
                return parents;
            }
            for (DAGNode<T> node : nodes.values()) {
                if (node.getChildren().contains(childNode)) {
                    parents.add(node.data);
                }
            }
            return parents;
        }

        public void breadthFirstTraversal(T start, Consumer<T> consumer) {
//            List<T> result = new ArrayList<>();

            DAGNode<T> startNode = nodes.get(start);
            if (startNode == null) {
                return;
            }

            Queue<DAGNode<T>> queue = new LinkedList<>();
            queue.add(startNode);

            while (!queue.isEmpty()) {
                DAGNode<T> current = queue.poll();
                consumer.accept(current.getData());
//                result.add(current.getData());

                queue.addAll(current.getChildren());
            }
        }
    }


    private static class DAGNode<T> {
        private final T data;
        private final List<DAGNode<T>> children;

        public DAGNode(T data) {
            this.data = data;
            this.children = new ArrayList<>();
        }

        public T getData() {
            return data;
        }

        public List<DAGNode<T>> getChildren() {
            return children;
        }

        public void addChild(DAGNode<T> child) {
            children.add(child);
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }

            DAGNode<?> dagNode = (DAGNode<?>) o;

            if (!Objects.equals(data, dagNode.data)) {
                return false;
            }
            return children.equals(dagNode.children);
        }

        @Override
        public int hashCode() {
            int result = data != null ? data.hashCode() : 0;
            result = 31 * result + children.hashCode();
            return result;
        }
    }

}
